package main

import (
	"fmt"
	"go-the-way-of-learning/day10/3log_agent_project/conf"
	"go-the-way-of-learning/day10/3log_agent_project/etcd"
	"go-the-way-of-learning/day10/3log_agent_project/kafka"
	"go-the-way-of-learning/day10/3log_agent_project/taillog"
	"go-the-way-of-learning/day10/3log_agent_project/util"
	"gopkg.in/ini.v1"
	"sync"
	"time"
)

// 加载配置文件
var cfg = new(conf.AppConf)

//func run() {
//	// 1、 读日志文件
//	for true {
//		select {
//		case line := <-taillog.ReadLogChan():
//			// 2、 发送kafka
//			kafka.SendKafka(cfg.KafkaConf.Topic, line.Text)
//		default:
//			time.Sleep(time.Second)
//		}
//	}
//}

func main() {
	err := ini.MapTo(cfg, "./day10/3log_agent_project/conf/config.ini")
	if err != nil {
		fmt.Printf("load ini failed, err: %v\n", err)
		return
	}
	// 初始化kafka连接
	err = kafka.Init([]string{cfg.KafkaConf.Address}, cfg.KafkaConf.ChanMaxSize)
	if err != nil {
		fmt.Printf("init kafka failed, err: %T\n", err)
		return
	}
	fmt.Println("kafka init success")

	// 初始化etcd
	err = etcd.Init(cfg.EtcdConf.Address, time.Duration(cfg.EtcdConf.Timeout)*time.Second)
	if err != nil {
		fmt.Printf("init etcd failed, err: %T\n", err)
		return
	}
	fmt.Println("etcd init success")

	// 1、从etcd获取日志收集项的配置信息
	ipStr, err := util.GetOutboundIP()
	if err != nil {
		panic(err)
	}
	etcdConfKey := fmt.Sprintf(cfg.EtcdConf.Key, ipStr)
	logEntryConf, err := etcd.GetConf(etcdConfKey)
	if err != nil {
		fmt.Printf("etcd.GetConf failed, err: %v\n", err)
		return
	}
	fmt.Printf("get conf from etcd success, %v\n", logEntryConf)
	// 2、 派一个哨兵去监视日志收集项的变化

	for index, value := range logEntryConf {
		fmt.Printf("index:%v value:%v\n", index, value)
	}
	// 3、 收集日志发往kafka
	// 因为taillog.NewConfChan()访问了tskMgr的newConfChan,所以先调用这个初始化方法
	taillog.Init(logEntryConf)

	newConfChan := taillog.NewConfChan()
	var wg sync.WaitGroup
	wg.Add(1)
	go etcd.WatchConf(etcdConfKey, newConfChan)
	wg.Wait()
	//run()
}
